package com.czxy

import java.util.Properties

import com.czxy.bean.{HBaseMeta, TagRule}
import org.apache.spark.sql._
import org.apache.spark.sql.expressions.UserDefinedFunction

/*
 * @Auther: Alice菌
 * @Date: 2020/6/4 15:26
 * @Description:
    流年笑掷 未来可期。以梦为马,不负韶华!
 */


/*  Gender 用于性别标签的计算  */
object GenderTags {
  // 程序的入口
  def main(args: Array[String]): Unit = {

    // 1. 创建SparkSQL             【1周文均】
    //    用于读取mysql ， hbase等数据
    val spark: SparkSession = SparkSession.builder().appName("GenderTag").master("local[*]").getOrCreate()

    // 设置日志级别
    spark.sparkContext.setLogLevel("WARN")

    //2  连接mysql 数据库     【2张辉】

    //url: String, table: String, properties: Properties
    // 设置Spark连接MySQL所需要的字段
    var url: String = "jdbc:mysql://bd001:3306/tags_new2?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&user=root&password=123456"
    var table: String = "tbl_basic_tag" //mysql数据表的表名
    var properties: Properties = new Properties

    //连接mysql
    val mysqlConn: DataFrame = spark.read.jdbc(url, table, properties)

    //引入隐式转换            【3熊颖悟】
    import spark.implicits._
    //引入java 和scala相互转换
    import scala.collection.JavaConverters._
    //引入sparkSQL的内置函数
    import org.apache.spark.sql.functions._


    //3 读取Mysql数据库的四级标签
    //  为读取hbase数据做准备
    val fourTagsDS: Dataset[Row] = mysqlConn.select("id", "rule").where("id=4")
    // 这个字符串读取数据不好用   inType=HBase##zkHosts=192.168.10.20##zkPort=2181##hbaseTable=tbl_users##family=detail##selectFields=id,gender
    // 将上述数据转为样例类，以便于后面读取数据
    // 遍历四级标签数据fourTags
    val KVMap: Map[String, String] = fourTagsDS.map(row => {

      // 获取到rule的值
      val RuleValue: String = row.getAs("rule").toString

      /*
           inType=HBase##
           zkHosts=192.168.10.20##
           zkPort=2181##
           hbaseTable=tbl_users##
           family=detail##
           selectFields=id,gender
        */

      //使用“##”对数据继续切分
      val KVMaps: Array[(String, String)] = RuleValue.split("##").map(kv => {
        val arr: Array[String] = kv.split("=")
        //zkHosts     192.168.10.20 , zkPort   2181
        (arr(0), arr(1))
      })
      KVMaps
    }).collectAsList().get(0).toMap
    // 考虑到KVMaps的返回值类型为Dataset[Array[(String, String)]]
    // 为了后续方便调用，我们这里引入隐式转换，使用collectAsList将其转换成List，再转换成的Map

    println(KVMap)
    /*
         Map(selectFields -> id,gender, inType -> HBase, zkHosts -> 192.168.10.20, zkPort -> 2181, hbaseTable -> tbl_users, family -> detail)
       */

    //开发toHBaseMeta方法     将KVMap 封装成为样例类HBaseMeta    【4杨贵东】
    var hbaseMeta: HBaseMeta = toHBaseMeta(KVMap)

    //println( hbaseMeta.hbaseTable+"    "+ hbaseMeta.family+"    "+ hbaseMeta.selectFields)
    /*
        tbl_users    detail    id,gender
      */
    //4  读取mysql数据库中的五级标签      【5胡熙梅】
    //   匹配性别
    val fiveTagsDS: Dataset[Row] = mysqlConn.select('id, 'rule).where("pid=4")
    // 获取出id 和 rule
    // 将FiveTagsDS  封装成样例类TagRule
    val fiveTageList: List[TagRule] = fiveTagsDS.map(row => {

      // row 是一条数据
      // 获取出id 和 rule
      val id: Int = row.getAs("id").toString.toInt
      val rule: String = row.getAs("rule").toString

      // 封装样例类
      TagRule(id, rule)
    }).collectAsList() // 将DataSet转换成util.List[TagRule]   这个类型遍历时无法获取id,rule数据
      .asScala.toList // 将util.List转换成list   需要隐式转换    import scala.collection.JavaConverters._

    //    for(a<- fiveTageList){
    //       println(a.id+"      "+a.rule)
    //    }

    //5      1
    //6      2

    // 5. 根据mysql数据中的四级标签， 读取hbase数据            【6杜友灝】
    // 若使用hbase 客户端读取效率较慢，将hbase作为【数据源】，读取效率较快

    val hbaseDatas: DataFrame = spark.read.format("com.czxy.tools.HBaseDataSource")
      // hbaseMeta.zkHosts 就是 192.168.10.20  和 下面是两种不同的写法
      .option("zkHosts", hbaseMeta.zkHosts)
      .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)
      .option(HBaseMeta.HBASETABLE, hbaseMeta.hbaseTable)
      .option(HBaseMeta.FAMILY, hbaseMeta.family)
      .option(HBaseMeta.SELECTFIELDS, hbaseMeta.selectFields)
      .load()

    // 展示一些数据
    hbaseDatas.show(5)
    /*
         +---+------+
         |  1|     2|
         | 10|     2|
         |100|     2|
         |101|     1|
         |102|     2|
       */
    // 6 标签匹配       【7张俊奇】
    //   根据五级标签数据和hbase数据进行标签匹配  得到最终的标签

    // 编写udf函数  例如输入是1,2   返回不同性别对应的id值5或者6
    val GetTagId: UserDefinedFunction = udf((gender: String) => {
      // 设置标签默认值
      var id: Int = 0
      // 遍历五级标签
      for (ruleOb <- fiveTageList) {
        // 当用户数据的gender与五级标签的id相等
        // 那么返回五级标签的id
        if (gender == ruleOb.rule) {
          id = ruleOb.id
        }
      }
      id
    })

    // 标签匹配
    val userTags: DataFrame = hbaseDatas.select('id.as("userId"), GetTagId('gender).as("tagsId"))

    // 输出查看效果
    userTags.show()

    /*
      +------+------+
      |userId|tagsId|
      +------+------+
      |     1|     6|
      |    10|     6|
      |   100|     6|

       */

    //7 将最终的标签写入Hbase      【8郭昕玮】
    userTags.write.format("com.czxy.tools.HBaseDataSource")
      .option("zkHosts", hbaseMeta.zkHosts)
      .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)
      .option(HBaseMeta.HBASETABLE, "test")
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS, "userId,tagsId")
      .save()


  }


  //将mysql中的四级标签的rule  封装成HBaseMeta
  //方便后续使用的时候方便调用
  def toHBaseMeta(KVMap: Map[String, String]): HBaseMeta = {
    //开始封装
    HBaseMeta(KVMap.getOrElse("inType", ""),
      KVMap.getOrElse("zkHosts", ""),
      KVMap.getOrElse("zkPort", ""),
      KVMap.getOrElse("hbaseTable", ""),
      KVMap.getOrElse("family", ""),
      KVMap.getOrElse("selectFields", ""),
      KVMap.getOrElse("rowKey", "")
    )
  }

}

